/*
 * @Author: wangjq4214
 * @Date: 2022-04-18 11:04:49
 * @LastEditTime: 2022-06-07 19:25:45
 * @LastEditors: wangjq4214
 */
package main

import (
	"context"
	"log"
	"sync"
	"sync/atomic"
	"time"

	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"

	pb "gitee.com/wangjq4214/schedule/proto/follower"
)

type Runner struct {
	mu sync.Mutex

	waitTask    map[string]*Task
	readyTask   []*Task
	runningTask map[string]*Task

	commitChan chan Task
	finishChan chan Task

	status *Status
}

func NewRunner(commitChan chan Task, finishChan chan Task, status *Status) *Runner {
	return &Runner{
		waitTask:    map[string]*Task{},
		readyTask:   []*Task{},
		runningTask: map[string]*Task{},
		commitChan:  commitChan,
		finishChan:  finishChan,
		status:      status,
	}
}

func (r *Runner) RunTask() {
	for {
		r.mu.Lock()
		// 没有任务休眠
		if len(r.readyTask) == 0 {
			r.mu.Unlock()
			time.Sleep(5 * time.Second)
			continue
		}

		// 任务无法运行休眠
		cpuExceed := r.readyTask[0].CPU > int(r.status.CPUTotal)-int(r.status.CPUUsing)
		storageExceed := r.readyTask[0].Storage > int(r.status.StorageTotal)-int(r.status.StorageUsing)
		if cpuExceed || storageExceed {
			r.mu.Unlock()
			time.Sleep(5 * time.Second)
			continue
		}

		// 运行任务
		// 更新节点状态
		item := r.readyTask[0]
		atomic.AddInt32(&r.status.CPUUsing, int32(item.CPU))
		atomic.AddInt32(&r.status.CPUWait, -int32(item.CPU))
		atomic.AddInt32(&r.status.StorageUsing, int32(item.Storage))
		// 加入运行集合
		r.readyTask = r.readyTask[1:]
		r.runningTask[item.UUID] = item

		go func(task *Task) {
			task.StartTime = time.Now().Unix()
			time.Sleep(time.Duration(task.Time) * time.Second)

			r.mu.Lock()

			atomic.AddInt32(&r.status.CPUUsing, -int32(task.CPU))
			atomic.AddInt32(&r.status.StorageUsing, -int32(task.Storage))

			delete(r.runningTask, task.UUID)
			task.FinishTime = time.Now().Unix()
			r.finishChan <- *task

			r.mu.Unlock()
		}(item)

		r.mu.Unlock()
	}
}

func (r *Runner) getData(uuid, addr string, id int32, cb func()) {
	// 无数据
	if addr == "" {
		r.mu.Lock()
		cb()
		r.mu.Unlock()
		return
	}

	// 请求数据
	conn, err := grpc.Dial(addr,
		grpc.WithTransportCredentials(insecure.NewCredentials()),
		grpc.WithDefaultCallOptions(
			grpc.MaxCallSendMsgSize(500*1024*1024),
			grpc.MaxCallRecvMsgSize(500*1024*1024),
		),
	)
	if err != nil {
		log.Printf("did not connect: %v", err)
		return
	}
	defer conn.Close()
	c := pb.NewFollowerServeClient(conn)

	// Contact the server and print out its response.
	ctx, cancel := context.WithTimeout(context.Background(), 60*60*time.Second)
	defer cancel()
	client, err := c.GetData(ctx, &pb.GetDataReq{
		ID: id,
	})
	if err != nil {
		log.Printf("could not send: %v", err)
		return
	}

	for {
		_, err := client.Recv()
		if err != nil {
			if err.Error() == "EOF" {
				break
			}

			log.Printf("Get data error: %v", err)
			break
		}
	}

	log.Println("Get data success")

	r.mu.Lock()
	cb()
	r.mu.Unlock()
}

func (r *Runner) SaveTask() {
	// 接收任务
	for {
		item := <-r.commitChan

		r.mu.Lock()

		item.ReceiveTime = time.Now().Unix()
		// 将任务添加到等待队列并且更新状态
		r.waitTask[item.UUID] = &item
		atomic.AddInt32(&r.status.CPUWait, int32(item.CPU))
		log.Printf("Receive task: %v\n", item.UUID)

		go r.getData(item.UUID, item.Position.IP, item.Position.ID, func() {
			tmp := r.waitTask[item.UUID]
			delete(r.waitTask, item.UUID)
			tmp.ReadyTime = time.Now().Unix()
			r.readyTask = append(r.readyTask, tmp)
		})

		r.mu.Unlock()
	}
}
